package com.lagou.processor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.filter.ByteArrayComparable;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.NavigableMap;

public class MyProcessor extends BaseRegionObserver {


    @Override
    public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> e,
                           final Delete delete, final WALEdit edit, final Durability durability)
            throws IOException {

        RegionCoprocessorEnvironment environment = e.getEnvironment();
        HTableInterface userRelation = environment.getTable(TableName.valueOf("user_relation"));

        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();// 获取删除的所有的列簇及列
        for (byte[] family : familyCellMap.keySet()) {
            List<Cell> cells = familyCellMap.get(family);// 获取列族下所有的列
            for (Cell cell : cells) {
                // 获取rowkey  1
                byte[] cloneRow = CellUtil.cloneRow(cell);
                System.out.println("已删除的rowkey:"+Bytes.toString(cloneRow));
                // 获取列  2
                byte[] qualifier = CellUtil.cloneQualifier(cell);

                // 判断将要删除的数据(2 fid 1)是否存在，存在即删除，不存在不处理
                boolean exists = userRelation.exists(new Get(qualifier).addColumn(Bytes.toBytes("friends"), cloneRow));
                if(exists){
                    Delete friends = new Delete(qualifier).addColumn(Bytes.toBytes("friends"), cloneRow);
                    userRelation.delete(friends);
                }
            }
        }
    }
}
